package com.mxd.flink.connector.operator;

import com.mxd.flink.connector.config.Delimiters;
import com.mxd.flink.connector.config.RedisOptions;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.flink.configuration.ReadableConfig;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisSentinelPool;
import java.util.Arrays;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author rongdi
 * @date 2022/9/19 17:45
 */
public class RedisOperators {

    /**
     * 根据配置项获取RedisOperator对象
     * @param config
     * @return
     */
    public static RedisOperator get(ReadableConfig config) {
        String redisMode = config.get(RedisOptions.REDISMODE);
        final GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
        genericObjectPoolConfig.setMaxIdle(config.get(RedisOptions.MAXIDLE));
        genericObjectPoolConfig.setMaxTotal(config.get(RedisOptions.MAXTOTAL));
        genericObjectPoolConfig.setMinIdle(config.get(RedisOptions.MINIDLE));
        if("single".equals(redisMode)) {
            final JedisPool jedisPool = new JedisPool(genericObjectPoolConfig, config.get(RedisOptions.HOST), config.get(RedisOptions.PORT), config.get(RedisOptions.TIMEOUT), config.get(RedisOptions.PASSWORD), config.get(RedisOptions.DATABASE));
            return new RedisSingleOperator(jedisPool);
        } else if("cluster".equals(redisMode)) {
            String nodeStr = config.get(RedisOptions.CLUSTERNODES);
            Set<HostAndPort> nodes = Arrays.asList(nodeStr.split(Delimiters.COMMA)).stream().map(r -> {
                String[] arr = r.split(Delimiters.COLON);
                return new HostAndPort(arr[0].trim(), Integer.parseInt(arr[1].trim()));
            }).collect(Collectors.toSet());
            final JedisCluster jedisCluster = new JedisCluster(nodes, config.get(RedisOptions.TIMEOUT), config.get(RedisOptions.TIMEOUT), 2, config.get(RedisOptions.PASSWORD), genericObjectPoolConfig);
            return new RedisClusterOperator(jedisCluster);
        } else if("sentinel".equals(redisMode)) {
            String sentinelsInfo = config.get(RedisOptions.SENTINELS_INFO);
            Set<String> sentinels = Arrays.asList(sentinelsInfo.split(Delimiters.COMMA)).stream().collect(Collectors.toSet());
            final JedisSentinelPool jedisSentinelPool = new JedisSentinelPool(config.get(RedisOptions.REDIS_MASTER_NAME), sentinels, genericObjectPoolConfig, config.get(RedisOptions.TIMEOUT), config.get(RedisOptions.TIMEOUT), config.get(RedisOptions.SENTINELS_PASSWORD), config.get(RedisOptions.DATABASE));
            return new RedisSingleOperator(jedisSentinelPool);
        }
        throw new IllegalArgumentException("Jedis configuration not found");
    }

    public static RedisOperator getSimple(String host,Integer port,String password,Integer database) {
        final GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
        final JedisPool jedisPool = new JedisPool(genericObjectPoolConfig, host, port, 4000, password, database);
        return new RedisSingleOperator(jedisPool);
    }

}
